package org.example.branch_tag;

import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class CreateTag {
    public static void main(String[] args) throws Exception {
        // 配置Flink Web UI端口和并行度
        Configuration config = new Configuration();
        config.setString("rest.bind-port", "8081"); // 设置Web UI端口为8081
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        env.setParallelism(1); // 设置并行度为1
        env.enableCheckpointing(5000); // 设置checkpoint interval为5秒

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建Paimon目录
        String createCatalogSQL = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        System.out.println("正在执行SQL: \n" + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        String useCatalogSQL = "USE CATALOG paimon;";
        System.out.println("正在执行SQL: \n" + useCatalogSQL);
        tableEnv.executeSql(useCatalogSQL);

        // 判断当前是否已经存在v1.5tag
        String checkTagSQL = "SELECT * FROM `paimon`.`default`.`LogTable$tags` WHERE tag_name = 'v1.5';";
        System.out.println("正在执行SQL: \n" + checkTagSQL);
        tableEnv.executeSql(checkTagSQL).print();
    
        // 如果存在，则删除
        CloseableIterator<Row> iterator = tableEnv.executeSql(checkTagSQL).collect();
        boolean tagExists = iterator.hasNext();
        iterator.close();
        
        if (tagExists) {
            String deleteTagSQL = "CALL sys.delete_tag('default.LogTable', 'v1.5');";
            System.out.println("正在执行SQL: \n" + deleteTagSQL);
            tableEnv.executeSql(deleteTagSQL);
        } else {
            System.out.println("v1.5 tag不存在，创建v1.5 tag");
        }

        // 为LogTable创建标签
        String createTagSQL = "CALL sys.create_tag('default.LogTable', 'v1.5');";
        System.out.println("正在执行SQL: \n" + createTagSQL);
        tableEnv.executeSql(createTagSQL);

        // 查询标签信息
        String showTagsSQL = "SELECT * FROM `paimon`.`default`.`LogTable$tags`";
        System.out.println("正在执行SQL: \n" + showTagsSQL);
        tableEnv.executeSql(showTagsSQL).print();

        // 查询分支信息
        String showBranchesSQL = "SELECT * FROM `paimon`.`default`.`LogTable$branches`";
        System.out.println("正在执行SQL: \n" + showBranchesSQL);
        tableEnv.executeSql(showBranchesSQL).print();
    }
}
